package com.pangyangyang.elasticsearch.plugin.dispatch;

import com.pangyangyang.elasticsearch.plugin.event.DataModifyEvent;
import com.pangyangyang.elasticsearch.plugin.exception.MessageRiverException;
import com.pangyangyang.elasticsearch.plugin.river.MessageRiver;

import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by yangwang on 2015/1/16.
 */
public class EventDispatcherHolder {

    private static EventDispatcher instance;
    private static AtomicInteger riverInstanceCount;
    private static boolean started = false;

    public static Future dispatchEvent(DataModifyEvent event) throws MessageRiverException{
        if (!started){
            throw new MessageRiverException("All dispatcher shutdown.");
        }
        return instance.dispatchEvent(event);
    }

    public static void registerRiver(MessageRiver river){
        if (!started){
            startDispatcher();
        }
        instance.registerHandler(river.getHandler());
        riverInstanceCount.incrementAndGet();
    }

    public static void unregisterRiver(MessageRiver river){
        instance.unregisterHandler(river.getDestination());
        if(riverInstanceCount.decrementAndGet() == 0){
            stopDispatcher();
        }
    }

    protected static synchronized void startDispatcher(){
        if(!started){
            riverInstanceCount = new AtomicInteger(0);
            instance = new EventDispatcher();
            started = true;
        }
    }

    protected static void stopDispatcher(){
        instance.stop();
        riverInstanceCount = null;
        started = false;
    }
}
